package edu.nf.llmtest.service.impl;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import edu.nf.llmtest.common.converter.ConversationsConverter;
import edu.nf.llmtest.common.converter.SessionConverter;
import edu.nf.llmtest.common.enumeration.RedisKeyEnum;
import edu.nf.llmtest.model.document.ConversationsDoc;
import edu.nf.llmtest.model.dto.EssayDTO;
import edu.nf.llmtest.model.entity.Conversations;
import edu.nf.llmtest.model.entity.Session;
import edu.nf.llmtest.service.Listenter.MqProducer;
import edu.nf.llmtest.service.LlmService;
import edu.nf.llmtest.service.TrafficService;
import edu.nf.llmtest.service.WeatherService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.elasticsearch.core.ElasticsearchOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.reactive.function.client.WebClientResponseException;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.atomic.AtomicReference;

/**
 * LLM服务实现类
 * 
 * <p>职责：统一封装对 SiliconFlow 大模型的调用（支持 Qwen、DeepSeek、GLM 等模型），
 * 提供流式对话、函数调用、会话管理等功能。</p>
 * 
 * <p>主要功能：</p>
 * <ul>
 *   <li>1. 流式多轮对话：支持上下文记忆，自动管理会话历史</li>
 *   <li>2. 函数调用：支持天气查询、火车班次查询等工具函数</li>
 *   <li>3. 会话管理：自动创建会话、生成标题、清空上下文</li>
 *   <li>4. 数据同步：将对话数据同步到 Redis、Elasticsearch 和 MySQL</li>
 * </ul>
 * 
 * <p>使用方式：在 Controller 中注入 {@code @Autowired LlmServiceImpl llmService}，
 * 调用 {@code llmService.callStream(sessionId, systemPrompt, userPrompt, userId)} 即可返回流式响应。</p>
 *
 * @author bin
 * @since 2025/9/28
 */
@Service
@RequiredArgsConstructor
@Slf4j
public class LlmServiceImpl implements LlmService {
    /** SiliconFlow API 的 WebClient 客户端 */
    private final WebClient siliconWebClient;
    
    /** 当前使用的模型名称（从配置文件读取） */
    @Value("${silicon.model}")
    private String model;

    /** 每次读取的上下文长度限制（最多保留最近N轮对话） */
    private static final int CONTEXT_LIMIT = 10;

    /** 函数调用最大递归深度限制（防止无限循环） */
    private static final int MAX_FUNCTION_CALL_DEPTH = 5;
    
    /** 用于去重的函数调用记录（key: sessionId, value: 已调用的函数签名集合） */
    private static final java.util.concurrent.ConcurrentHashMap<String, java.util.Set<String>> functionCallHistory = new java.util.concurrent.ConcurrentHashMap<>();

    /** 交通服务：用于查询火车班次信息 */
    private final TrafficService trafficService;

    /** 天气服务：用于查询天气信息 */
    private final WeatherService weatherService;

    /** Redis 模板：用于存储会话和对话历史 */
    private final RedisTemplate<String, Object> redisTemplate;

    /** 会话转换器：用于 Session 实体与文档之间的转换 */
    private final SessionConverter sessionConverter;

    /** 对话转换器：用于 Conversations 实体与文档之间的转换 */
    private final ConversationsConverter converter;

    /** Elasticsearch 操作：用于全文搜索 */
    private final ElasticsearchOperations operations;

    /** 消息队列生产者：用于异步同步数据到数据库 */
    private final MqProducer mqProducer;

    /** JSON对象转换器 */
    private final ObjectMapper mapper;

    /**向量数据库操作*/
    private final QdrantVectorService qdrantService;

    /**
     * 流式多轮对话接口
     * 
     * <p>功能说明：</p>
     * <ul>
     *   <li>支持多轮对话上下文记忆（最多保留最近10轮）</li>
     *   <li>支持流式输出，实时返回AI回复内容</li>
     *   <li>支持函数调用（天气查询、火车班次查询）</li>
     *   <li>自动管理会话：首次对话时创建Session，对话结束后保存到Redis和ES</li>
     * </ul>
     * 
     * <p>工作流程：</p>
     * <ol>
     *   <li>从Redis获取历史对话记录（最多CONTEXT_LIMIT条）</li>
     *   <li>如果是首次对话，创建新的Session</li>
     *   <li>构建消息列表（包含历史对话和当前用户输入）</li>
     *   <li>构造LLM请求体（包含工具函数定义）</li>
     *   <li>发送流式请求，实时解析响应</li>
     *   <li>检测函数调用：如果AI请求调用函数，执行函数并再次请求</li>
     *   <li>流式返回内容：直接返回文本内容，不包装JSON</li>
     *   <li>对话结束后保存AI回复到Redis和ES</li>
     * </ol>
     * 
     * @param sessionId 会话ID（每个用户、每个tab、每个聊天框都有唯一ID）
     * @param system 系统提示词（定义AI的角色和行为规则）
     * @param user 用户输入的消息内容
     * @param userId 用户ID
     * @return 流式响应，直接返回文本内容片段（不是JSON格式）
     */
    @Override
    public Flux<String> callStream(String sessionId, String system, String user, Integer userId) {
        // ========== 第一步：获取历史对话上下文 ==========
        // 从Redis中获取该会话的所有历史对话记录
        List<Object> history = redisTemplate.opsForList().range(
                RedisKeyEnum.CONVERSATION_PERFIX.getValue() + sessionId, 0, -1);

        // ========== 第二步：创建新会话（仅在首次对话时） ==========
        // 判断依据：如果该sessionId没有历史记录，说明是首次对话
        if (history == null || history.isEmpty()) {
            Session session = new Session(sessionId, userId, "生成中", LocalDateTime.now().toString());
            // 将会话添加到用户的会话列表中
            redisTemplate.opsForList().rightPush(RedisKeyEnum.SESSIN_PERFIX.getValue() + userId, session);
            log.info("✅ 创建新会话 - sessionId={}, userId={}", sessionId, userId);
        }

        // ========== 第三步：构建消息列表（用于发送给LLM） ==========
        List<Map<String, String>> messages = new ArrayList<>();

        // 首先添加系统提示词（必须在消息列表的开头）
        messages.add(Map.of("role", "system", "content", system));

        // 如果有历史对话，加载最近的对话记录（最多CONTEXT_LIMIT条）
        if (history != null && !history.isEmpty()) {
            int startIndex = Math.max(0, history.size() - CONTEXT_LIMIT);
            for (int i = startIndex; i < history.size(); i++) {
                Conversations conv = (Conversations) history.get(i);
                if (!"system".equals(conv.getRole())) {
                    String message = conv.getMessage();
                    // 🚨 过滤掉错误的函数调用文本（避免AI学习错误模式）
                    if (message != null && 
                        (message.contains("ロン") || 
                         message.contains("<tool_call>") ||
                         (message.startsWith("{") && message.contains("\"name\"") && message.contains("\"arguments\"") && message.contains("query_")))) {
                        log.warn("⚠️ 过滤掉历史对话中的错误函数调用文本: {}", 
                                message.length() > 50 ? message.substring(0, 50) + "..." : message);
                        continue; // 跳过这条错误记录
                    }
                    messages.add(Map.of("role", "assistant", "content", message));
                }
            }
        }

        // ========== 第四步：添加当前用户输入 ==========
        // 将用户消息添加到消息列表
        messages.add(Map.of("role", "user", "content", user));
        // 保存用户消息到Redis和ES
        addConversation(sessionId, userId, "user", user);

        // ========== 第六步：定义工具函数（供AI调用） ==========
        // 获取火车班次查询函数的定义
        Map<String, Object> trafficFunction = TrafficServiceImpl.functionSchema();
        // 获取天气查询函数的定义
        Map<String, Object> weatherFunction = WeatherServiceImpl.functionSchema();

        // 获取向量查询函数的定义
        Map<String, Object> vectorFunction = QdrantVectorService.functionSchema();

        // ========== 第七步：构建LLM请求体 ==========
        // 根据模型类型选择 functions 或 tools 格式
        Map<String, Object> bodyMap = new HashMap<>();
        bodyMap.put("model", model);
        bodyMap.put("temperature", 0.3);
        bodyMap.put("stream", true);
        bodyMap.put("messages", messages);
        
        // 判断模型类型，选择对应的函数调用格式
        if (model != null && model.contains("Qwen2-7B")) {
            // Qwen2-7B-Instruct 使用 functions 格式
            List<Map<String, Object>> functions = List.of(trafficFunction, weatherFunction, vectorFunction);
            bodyMap.put("functions", functions);
            bodyMap.put("function_call", "auto");
        } else {
            // Qwen2.5 或其他模型使用 tools 格式
            List<Map<String, Object>> tools = List.of(
                Map.of("type", "function", "function", trafficFunction),
                Map.of("type", "function", "function", weatherFunction),
                Map.of("type", "function", "function", vectorFunction)
            );
            bodyMap.put("tools", tools);
            bodyMap.put("tool_choice", "auto");
        }

        StringBuilder aiBuffer = new StringBuilder();
        AtomicReference<String> functionNameRef = new AtomicReference<>("");
        AtomicReference<String> functionArgsRef = new AtomicReference<>("");
        AtomicReference<String> toolCallIdRef = new AtomicReference<>("");
        AtomicReference<Boolean> shouldIgnoreRef = new AtomicReference<>(false);  // 标记是否应该忽略后续数据

        // 验证system消息是否已添加到请求体
        log.info("📤 发送LLM请求，消息数量: {}, 模型: {}", messages.size(), model);
        log.info("📤 函数列表: query_train_schedule, query_weather, query_qdrant");
        boolean hasSystem = messages.stream()
                .anyMatch(msg -> "system".equals(msg.get("role")));
        if (!hasSystem) {
            log.error("❌ 警告：请求体中缺少system消息！");
        }
        
        // 打印完整请求体（调试用）
        try {
            log.debug("📤 完整请求体: {}", mapper.writeValueAsString(bodyMap));
        } catch (Exception e) {
            log.warn("无法序列化请求体");
        }

        // ========== 发送流式请求并处理响应 ==========
        return siliconWebClient.post()
                .uri("/chat/completions")
                .bodyValue(bodyMap)
                .retrieve()
                // 将响应转换为流式数据流
                .bodyToFlux(String.class)
                // 处理每个数据块
                .flatMap(chunk -> {
                    try {
                        // 解析SSE格式的数据（格式：data: {...}）
                        String data = chunk.trim();
                        if (data.startsWith("data:")) data = data.substring(5).trim();
                        // 跳过空数据和结束标记
                        if (data.isEmpty() || "[DONE]".equals(data)) return Flux.empty();

                        // 🔍 调试：打印每个响应chunk
                        log.debug("📥 收到chunk: {}", data.length() > 200 ? data.substring(0, 200) + "..." : data);

                        JsonNode root = mapper.readTree(data);
                        JsonNode choice = root.path("choices").get(0);
                        String finishReason = choice.path("finish_reason").asText("");
                        
                        // 🔍 调试：检测finish_reason
                        if (!finishReason.isEmpty()) {
                            log.info("🏁 finish_reason: {}", finishReason);
                        }
                        
                        // 检测函数调用（支持两种格式）
                        // 1. Qwen2-7B 使用 function_call 格式
                        JsonNode functionCall = choice.path("delta").path("function_call");
                        if (functionCall.isMissingNode()) {
                            functionCall = choice.path("message").path("function_call");
                        }
                        
                        // 2. Qwen2.5 使用 tool_calls 格式
                        JsonNode toolCalls = choice.path("delta").path("tool_calls");
                        if (toolCalls.isMissingNode()) {
                            toolCalls = choice.path("message").path("tool_calls");
                        }
                        
                        // 处理 tool_calls 格式（Qwen2.5）- 支持流式累积
                        if (!toolCalls.isMissingNode() && toolCalls.isArray() && toolCalls.size() > 0) {
                            log.debug("🔧 检测到 tool_calls 字段");
                            JsonNode toolCall = toolCalls.get(0);
                            
                            // 累积 tool_call_id
                            JsonNode idNode = toolCall.path("id");
                            if (!idNode.isMissingNode() && !idNode.isNull()) {
                                String idChunk = idNode.asText("");
                                if (!idChunk.isEmpty()) {
                                    toolCallIdRef.set(idChunk);
                                }
                            }
                            
                            // 累积函数信息
                            JsonNode functionNode = toolCall.path("function");
                            if (!functionNode.isMissingNode()) {
                                JsonNode nameNode = functionNode.path("name");
                                if (!nameNode.isMissingNode() && !nameNode.isNull()) {
                                    String nameChunk = nameNode.asText("");
                                    if (!nameChunk.isEmpty()) {
                                        log.info("🔧 tool_calls - function name chunk: {}", nameChunk);
                                        
                                        // ⚠️ 关键修复：检测是否是新的函数调用
                                        String currentName = functionNameRef.get();
                                        
                                        // 如果当前已经有完整的函数名，且新的chunk看起来是新函数的开始
                                        boolean isNewFunction = !currentName.isEmpty() && 
                                                (currentName.equals("query_train_schedule") || 
                                                 currentName.equals("query_weather") || 
                                                 currentName.equals("query_qdrant")) &&
                                                nameChunk.startsWith("query");
                                        
                                        if (isNewFunction) {
                                            log.info("⚠️ 检测到多个函数调用，保留第一个: {}, 忽略后续", currentName);
                                            shouldIgnoreRef.set(true);  // 标记为忽略模式
                                        } else if (!shouldIgnoreRef.get()) {
                                            functionNameRef.updateAndGet(current -> current + nameChunk);  // 追加
                                        }
                                    }
                                }
                                
                                // 只有在非忽略模式下才累积参数
                                if (!shouldIgnoreRef.get()) {
                                    JsonNode argsNode = functionNode.path("arguments");
                                    if (!argsNode.isMissingNode() && !argsNode.isNull()) {
                                        String argsChunk = argsNode.asText("");
                                        if (!argsChunk.isEmpty()) {
                                            log.debug("🔧 tool_calls - function args chunk: {}", argsChunk);
                                            functionArgsRef.updateAndGet(current -> current + argsChunk);
                                        }
                                    }
                                }
                            }
                        }
                        
                        // ⚠️ 关键修复：finish_reason 检查移到外层！
                        // 函数调用完成，执行函数（Qwen2.5 使用 tool_calls）
                        if ("tool_calls".equals(finishReason)) {
                            String name = functionNameRef.get();
                            String arguments = functionArgsRef.get();
                            String toolCallId = toolCallIdRef.get();
                            
                            log.info("✅ 函数调用完成 - name: {}, args: {}", name, arguments);
                            
                            if (!name.isEmpty()) {
                                functionNameRef.set("");
                                functionArgsRef.set("");
                                toolCallIdRef.set("");
                                shouldIgnoreRef.set(false);  // 重置忽略标志
                                log.info("🚀 开始执行函数调用: {}", name);
                                // 将 List<Map<String, String>> 转换为 List<Map<String, Object>>
                                List<Map<String, Object>> messagesObj = new ArrayList<>();
                                for (Map<String, String> msg : messages) {
                                    messagesObj.add(new HashMap<>(msg));
                                }
                                return handleFunctionCall(name, arguments, toolCallId, messagesObj, aiBuffer, sessionId, userId, 1);
                            } else {
                                log.warn("⚠️ finish_reason=tool_calls 但函数名为空！");
                                return Flux.empty();
                            }
                        }
                        
                        // 处理 function_call 格式（Qwen2-7B）- 支持流式累积
                        if (!functionCall.isMissingNode()) {
                            log.debug("🔧 检测到 function_call 字段");
                            
                            JsonNode nameNode = functionCall.path("name");
                            if (!nameNode.isMissingNode() && !nameNode.isNull()) {
                                String nameChunk = nameNode.asText("");
                                if (!nameChunk.isEmpty()) {
                                    log.info("🔧 function_call - name: {}", nameChunk);
                                    functionNameRef.updateAndGet(current -> current + nameChunk);
                                }
                            }
                            
                            JsonNode argsNode = functionCall.path("arguments");
                            if (!argsNode.isMissingNode() && !argsNode.isNull()) {
                                String argsChunk = argsNode.asText("");
                                if (!argsChunk.isEmpty()) {
                                    log.debug("🔧 function_call - args chunk: {}", argsChunk);
                                    functionArgsRef.updateAndGet(current -> current + argsChunk);
                                }
                            }
                            
                            // 函数调用完成，执行函数
                            if ("function_call".equals(finishReason)) {
                                String name = functionNameRef.get();
                                String arguments = functionArgsRef.get();
                                
                                log.info("✅ 函数调用完成 - name: {}, args: {}", name, arguments);
                                
                                if (!name.isEmpty()) {
                                    functionNameRef.set("");
                                    functionArgsRef.set("");
                                    log.info("🚀 开始执行函数调用: {}", name);
                                    // 将 List<Map<String, String>> 转换为 List<Map<String, Object>>
                                    List<Map<String, Object>> messagesObj = new ArrayList<>();
                                    for (Map<String, String> msg : messages) {
                                        messagesObj.add(new HashMap<>(msg));
                                    }
                                    return handleFunctionCall(name, arguments, "", messagesObj, aiBuffer, sessionId, userId, 1);
                                } else {
                                    log.warn("⚠️ finish_reason=function_call 但函数名为空！");
                                }
                            }
                            return Flux.empty();
                        }

                        // 正常流式内容输出（JSON格式）
                        JsonNode contentNode = choice.path("delta").path("content");
                        if (contentNode.isMissingNode() || contentNode.isNull()) return Flux.empty();
                        
                        String content = contentNode.isNumber() 
                                ? contentNode.asText() 
                                : contentNode.asText("");
                        
                        if (content.isEmpty() || "null".equals(content)) return Flux.empty();
                        
                        // 🔍 调试：打印AI输出的内容
                        if (content.length() > 0) {
                            log.debug("💬 AI输出: {}", content.length() > 50 ? content.substring(0, 50) + "..." : content);
                        }
                        
                        // 🚨 过滤错误的函数调用文本输出（单chunk检测）
                        if (content.contains("ロン") || 
                            content.contains("<tool_call>") || 
                            content.contains("</tool_call>") ||
                            content.contains("send{") ||
                            (content.contains("\"name\"") && content.contains("query_"))) {
                            log.warn("⚠️ 检测到AI错误地在content中输出函数调用标记，已过滤: {}", content);
                            return Flux.empty();
                        }
                        
                        // 累积到buffer
                        aiBuffer.append(content);
                        
                        // 🚨 累积检测：检查buffer中是否出现函数调用格式
                        String accumulated = aiBuffer.toString();
                        if (accumulated.length() > 10 && 
                            (accumulated.contains("ロン") || 
                             accumulated.contains("<tool_call>") ||
                             accumulated.contains("</tool_call>") ||
                             accumulated.contains("send{") ||
                             (accumulated.contains("{\"name\"") && accumulated.contains("\"arguments\"") && accumulated.contains("query_")))) {
                            log.warn("⚠️ 检测到AI在content中累积输出函数调用文本，已过滤: {}", 
                                    accumulated.length() > 100 ? accumulated.substring(0, 100) + "..." : accumulated);
                            aiBuffer.setLength(0); // 清空buffer
                            return Flux.empty();
                        }
                        
                        // 直接返回文本内容（SSE格式）
                        return Flux.just(content);

                    } catch (Exception e) {
                        // 解析错误时记录日志但不中断流
                        log.warn("解析流出错: {}", e.getMessage());
                        return Flux.empty();
                    }
                })
                // ========== 流式响应完成后的处理 ==========
                .doOnComplete(() -> {
                    // 流式输出完成后，保存AI的完整回复到Redis和ES
                    String aiMsg = aiBuffer.toString();
                    log.info("🔍 [主流程] doOnComplete 被触发 - aiBuffer长度={}", aiBuffer.length());
                    log.info("🔍 [主流程] AI回复内容: {}", aiMsg.length() > 200 ? aiMsg.substring(0, 200) + "..." : aiMsg);
                    if (!aiMsg.isEmpty()) {
                        // 🚨 过滤掉错误的输出格式（如：ロン{...} 或 {"name":"query_xxx"...}）
                        // 这些是AI模仿函数调用的纯文本输出，不应该保存到历史记录
                        if (aiMsg.contains("ロン") || 
                            aiMsg.contains("<tool_call>") ||
                            aiMsg.contains("</tool_call>") ||
                            aiMsg.contains("send{") ||
                            (aiMsg.contains("{\"name\"") && aiMsg.contains("\"arguments\"") && aiMsg.contains("query_"))) {
                            log.warn("⚠️ 检测到错误的AI输出格式（函数调用文本），不保存到历史: {}", 
                                    aiMsg.length() > 100 ? aiMsg.substring(0, 100) + "..." : aiMsg);
                        } else {
                            addConversation(sessionId, userId, "assistant", aiMsg);
                        }
                    }
                });
    }



    /**
     * 生成会话标题
     * 
     * @param content 用户的第一条消息内容（用于生成标题）
     * @param userId 用户ID
     * @return 包含标题信息的JSON字符串
     */
    @Override
    public Mono<String> generateTitle(String content,Integer userId) {
        // 检查最后一个会话是否已有标题（避免重复生成）
        String key = RedisKeyEnum.SESSIN_PERFIX.getValue() + userId;
        List<Object> sessions = redisTemplate.opsForList().range(key, -1, -1);
        
        if (sessions != null && !sessions.isEmpty()) {
            Object lastObj = sessions.get(0);
            if (lastObj instanceof Session session) {
                String existingTitle = session.getTitle();
                // 如果标题已存在且不是"生成中"，说明标题已生成，直接返回
                if (existingTitle != null && !"生成中".equals(existingTitle)) {
                    log.info("⚠️ 会话已有标题，跳过重复生成 - userId={}, title={}", userId, existingTitle);
                    return Mono.just("{\"message\": \"标题已存在\", \"title\": \"" + existingTitle + "\"}");
                }
            }
        }
        
        // 构建标题生成的提示词
        String systemPrompt = """
            你是一个智能助手，请从以下用户的消息中提炼出一个简短的会话标题（不超过10个字）。
            用户消息：{userMessage}
            例如："我想去重庆旅游" → "重庆旅游"
            不要解释，返回简短标题。
            """;

        Map<String, Object> body = Map.of(
                "model", model,
                "temperature", 0.3,
                "messages", List.of(
                        Map.of("role", "system", "content", systemPrompt),
                        Map.of("role", "user", "content", content)
                )
        );

        // 返回 Mono<String>，无需阻塞
        return siliconWebClient.post()
                .uri("/chat/completions")
                .bodyValue(body)
                .retrieve()
                .bodyToMono(String.class)
                .onErrorResume(WebClientResponseException.class, ex -> 
                    Mono.error(new RuntimeException("SiliconFlow API 调用失败: " + ex.getStatusCode(), ex))
                )
                .doOnSuccess(json -> {
                    try {
                        JsonNode root = mapper.readTree(json);
                        String title = root.path("choices").get(0)
                                .path("message")
                                .path("content")
                                .asText("");
                        if (!title.isBlank()) {
                            updateSessionTitle(userId, title);
                        }
                    } catch (Exception e) {
                        log.error("解析标题失败: {}", e.getMessage());
                    }
                });

    }

    /**
     * 清空某个会话的上下文
     * @param sessionId 会话ID
     */
    public void clearSession(String sessionId) {
        redisTemplate.delete(RedisKeyEnum.CONVERSATION_PERFIX.getValue() + sessionId);
        // 同时清空函数调用历史
        functionCallHistory.remove(sessionId);
        log.info("✅ 已清空会话 {} 的上下文和函数调用历史", sessionId);
    }

    /**
     * 更新会话标题
     * 
     * <p>功能说明：更新用户最后一个会话的标题，并同步到Elasticsearch和MySQL。</p>
     * 
     * <p>工作流程：</p>
     * <ol>
     *   <li>从Redis获取用户的会话列表</li>
     *   <li>找到最后一个会话（最新创建的）</li>
     *   <li>更新标题</li>
     *   <li>同步到Elasticsearch（用于搜索）</li>
     *   <li>通过MQ异步同步到MySQL</li>
     * </ol>
     * 
     * @param userId 用户ID
     * @param title 新的标题
     */
    public void updateSessionTitle(Integer userId, String title) {
        String key = RedisKeyEnum.SESSIN_PERFIX.getValue() + userId;
        List<Object> sessions = redisTemplate.opsForList().range(key, 0, -1);
        if (sessions == null || sessions.isEmpty()) return;

        Object lastObj = sessions.get(sessions.size() - 1);
        if (!(lastObj instanceof Session session)) return;

        session.setTitle(title);
        redisTemplate.opsForList().rightPop(key);
        redisTemplate.opsForList().rightPush(key, session);
        operations.save(sessionConverter.convertSessionDoc(session));
        mqProducer.send("addSession", session);
    }



    /**
     * 添加对话记录
     * @param sessionId 会话ID
     * @param userId 用户ID
     * @param role 角色（user、assistant、system等）
     * @param message 消息内容
     */
    public void addConversation(String sessionId, Integer userId, String role, String message) {
        log.info("📝 addConversation 被调用 - sessionId={}, userId={}, role={}, 消息长度={}", 
                sessionId, userId, role, message.length());
        log.info("📝 消息内容预览: {}", message.length() > 100 ? message.substring(0, 100) + "..." : message);
        
        Conversations conversations = new Conversations();
        // 先用UUID代替自增ID，在MQ中处理时返回真实的ID
        conversations.setMessageId(UUID.randomUUID().toString());
        conversations.setSessionId(sessionId);
        conversations.setUserId(userId);
        conversations.setRole(role);
        // 注意：emoji过滤在AddConversationsHanlder中处理，这里直接保存原始消息
        // 这样可以保留emoji在Redis和Elasticsearch中，只在MySQL中过滤
        conversations.setMessage(message);
        conversations.setCreatedTime(LocalDateTime.now().toString());

        // 将会话存进Redis中以list形式（用于上下文管理）
        redisTemplate.opsForList().rightPush(RedisKeyEnum.CONVERSATION_PERFIX.getValue() + sessionId, conversations);
        log.info("✅ 已保存到 Redis - key={}", RedisKeyEnum.CONVERSATION_PERFIX.getValue() + sessionId);

        // 同步到Elasticsearch（用于全文搜索）
        ConversationsDoc conversationsDoc = converter.convertConversationsDoc(conversations);
        operations.save(conversationsDoc);
        log.info("✅ 已保存到 Elasticsearch");

        // 通过MQ异步保存到MySQL（在AddConversationsHanlder中处理emoji过滤）
        mqProducer.send("addConversationHanlder", conversations);
        log.info("✅ 已发送到 MQ 队列");
    }

    /**
     * 处理函数调用（支持递归调用）
     * @param name 函数名称
     * @param arguments 函数参数（JSON字符串）
     * @param toolCallId 工具调用ID（用于关联函数调用和结果）
     * @param messages 当前的消息历史
     * @param aiBuffer AI回复缓冲区（用于累积完整回复）
     * @param sessionId 会话ID
     * @param userId 用户ID
     * @param depth 当前递归深度（用于防止无限循环）
     * @return 流式响应，返回AI基于函数结果生成的回复
     */
    private Flux<String> handleFunctionCall(String name, String arguments, String toolCallId, 
                                            List<Map<String, Object>> messages, 
                                             StringBuilder aiBuffer,
                                            String sessionId, Integer userId, int depth) {
        log.info("═══════════════════════════════════════════");
        log.info("🎯 进入 handleFunctionCall 方法");
        log.info("📌 函数名: {}", name);
        log.info("📌 参数: {}", arguments);
        log.info("📌 toolCallId: {}", toolCallId);
        log.info("📌 递归深度: {}/{}", depth, MAX_FUNCTION_CALL_DEPTH);
        log.info("═══════════════════════════════════════════");
        
        // ========== 检查递归深度限制 ==========
        if (depth > MAX_FUNCTION_CALL_DEPTH) {
            log.warn("⚠️ 函数调用递归深度超过限制({})，停止递归", MAX_FUNCTION_CALL_DEPTH);
            return Flux.just("函数调用次数过多，已达到最大递归深度限制。请基于已有信息直接为用户提供建议。");
        }
        
        // ========== 清理参数中的转义字符 ==========
        // 修复递归调用时参数被错误转义的问题
        if (arguments != null && arguments.contains("\\")) {
            log.debug("⚠️ 检测到参数中包含转义字符，原始参数: {}", arguments);
            String original = arguments;
            
            // 1. 移除可能的外层引号
            arguments = arguments.trim();
            if (arguments.startsWith("\"") && arguments.endsWith("\"")) {
                arguments = arguments.substring(1, arguments.length() - 1);
            }
            
            // 2. 移除末尾多余的 }"} 这种错误格式
            if (arguments.endsWith("\"}")) {
                arguments = arguments.substring(0, arguments.length() - 1);
            }
            
            // 3. 将转义的引号还原
            arguments = arguments.replace("\\\"", "\"");
            
            // 4. 验证是否是有效的JSON
            try {
                mapper.readTree(arguments);
                log.debug("✅ 清理后的参数: {}", arguments);
            } catch (Exception e) {
                log.warn("⚠️ 清理后的参数仍然无效，使用原始参数: {}", original);
                arguments = original;
            }
        }
        
        // ========== 检查函数调用去重 ==========
        String callSignature = name + "|" + arguments;
        java.util.Set<String> history = functionCallHistory.computeIfAbsent(sessionId, k -> java.util.concurrent.ConcurrentHashMap.newKeySet());
        if (history.contains(callSignature)) {
            log.warn("⚠️ 检测到重复的函数调用 - name: {}, args: {}", name, arguments);
            log.warn("⚠️ 该函数已在本次会话中被调用过，跳过执行");
            return Flux.just("该查询已执行过，请不要重复调用相同的函数。请继续执行其他查询（如天气查询、文章查询等），或基于已有信息直接为用户提供建议。");
        }
        history.add(callSignature);
        log.info("✅ 函数调用记录已添加 - 会话共调用 {} 个不同函数", history.size());
        
        try {
            // ========== 处理火车班次查询 ==========
            if ("query_train_schedule".equals(name)) {
                log.info("🚄 处理火车班次查询...");
                // 解析函数参数
                JsonNode args = mapper.readTree(arguments);
                String departure_station = args.path("departure_station").asText("");
                String arrival_station = args.path("arrival_station").asText("");
                String date = args.path("date").asText("");
                String filter = args.path("filter").asText("");
                String enable_booking = args.path("enable_booking").asText("1");
                String departure_time_range = args.path("departure_time_range").asText("");

                log.info("🧠 AI 请求调用函数：query_train_schedule({}, {}, {})",
                        departure_station, arrival_station, date);

                // 调用交通服务查询火车班次
                String resultString;
                boolean isApiError = false;
                try {
                    log.info("⏳ 正在调用本地函数 trafficService.getTrafficInfo()...");
                    resultString = trafficService
                            .getTrafficInfo(departure_station, arrival_station, date,
                                    filter, enable_booking, departure_time_range)
                            .toString();
                    log.info("✅ 火车班次查询成功！结果长度: {} 字符", resultString.length());
                    log.debug("📊 查询结果: {}", resultString.length() > 200 ? resultString.substring(0, 200) + "..." : resultString);
                    
                    // 检查是否是API错误（超出次数限制、请求频率限制等）
                    // 🔧 修复：检查 error_code 的值，而不是仅检查是否包含该字段
                    if (resultString.contains("\"error_code\": 200") || 
                        resultString.contains("\"error_code\":200") ||
                        resultString.contains("超过") || 
                        resultString.contains("限制") ||
                        resultString.contains("\"reason\": \"error\"") ||
                        resultString.contains("\"reason\":\"error\"")) {
                        isApiError = true;
                        log.warn("⚠️ API返回错误，将告知AI跳过此步骤");
                        resultString = "火车班次查询API暂时不可用（超出调用次数限制或请求频率限制），请跳过火车查询，继续其他步骤（如天气查询、文章查询等），并基于现有信息为用户提供建议。";
                    }
                } catch (Exception e) {
                    log.error("❌ 调用trafficService失败: {}", e.getMessage(), e);
                    isApiError = true;
                    resultString = "火车班次查询服务异常，请跳过此步骤，继续其他查询。";
                }

                // ========== 构建完整的消息历史（包括函数调用和结果） ==========
                List<Map<String, Object>> newMessages = new ArrayList<>();
                // 复制历史消息
                for (Map<String, Object> msg : messages) {
                    newMessages.add(new HashMap<>(msg));
                }
                // 添加assistant的函数调用请求（根据模型类型选择格式）
                Map<String, Object> assistantMsg = new HashMap<>();
                assistantMsg.put("role", "assistant");
                
                if (model != null && model.contains("Qwen2-7B")) {
                    // Qwen2-7B 使用 function_call 格式
                    assistantMsg.put("function_call", Map.of(
                            "name", name,
                            "arguments", arguments
                    ));
                } else {
                    // Qwen2.5 使用 tool_calls 格式
                    Map<String, Object> toolCall = new HashMap<>();
                    toolCall.put("id", toolCallId.isEmpty() ? "call_" + System.currentTimeMillis() : toolCallId);
                    toolCall.put("type", "function");
                    toolCall.put("function", Map.of("name", name, "arguments", arguments));
                    assistantMsg.put("tool_calls", List.of(toolCall));
                }
                newMessages.add(assistantMsg);
                
                // 添加函数执行结果（根据模型类型选择格式）
                Map<String, Object> toolResult = new HashMap<>();
                if (model != null && model.contains("Qwen2-7B")) {
                    // Qwen2-7B 使用 function 角色
                    toolResult.put("role", "function");
                    toolResult.put("name", name);
                    toolResult.put("content", resultString);
                } else {
                    // Qwen2.5 使用 tool 角色
                    toolResult.put("role", "tool");
                    toolResult.put("content", resultString);
                    if (!toolCallId.isEmpty()) {
                        toolResult.put("tool_call_id", toolCallId);
                    }
                }
                newMessages.add(toolResult);

                // 第二次请求LLM（根据模型类型选择格式）
                // ⚠️ 必须包含所有3个函数，让AI可以继续调用其他函数
                Map<String, Object> trafficFunc = TrafficServiceImpl.functionSchema();
                Map<String, Object> weatherFunc = WeatherServiceImpl.functionSchema();
                Map<String, Object> vectorFunc = QdrantVectorService.functionSchema();
                Map<String, Object> secondBody = new HashMap<>();
                secondBody.put("model", model);
                secondBody.put("stream", true);
                secondBody.put("messages", newMessages);
                
                if (model != null && model.contains("Qwen2-7B")) {
                    // Qwen2-7B 使用 functions 格式
                    List<Map<String, Object>> functions = List.of(trafficFunc, weatherFunc, vectorFunc);
                    secondBody.put("functions", functions);
                    secondBody.put("function_call", "auto");
                } else {
                    // Qwen2.5 使用 tools 格式
                    List<Map<String, Object>> toolsForSecond = List.of(
                            Map.of("type", "function", "function", trafficFunc),
                            Map.of("type", "function", "function", weatherFunc),
                            Map.of("type", "function", "function", vectorFunc)
                    );
                    secondBody.put("tools", toolsForSecond);
                    secondBody.put("tool_choice", "auto");
                }

                // 发送第二次请求，让AI基于函数结果生成回复
                AtomicReference<String> nextFunctionNameRef = new AtomicReference<>("");
                AtomicReference<String> nextFunctionArgsRef = new AtomicReference<>("");
                AtomicReference<String> nextToolCallIdRef = new AtomicReference<>("");
                
                return siliconWebClient.post()
                        .uri("/chat/completions")
                        .bodyValue(secondBody)
                        .retrieve()
                        .onStatus(status -> status.is4xxClientError() || status.is5xxServerError(), response -> {
                            return response.bodyToMono(String.class)
                                    .flatMap(errorBody -> {
                                        log.error("❌ 第二次API请求失败 - 状态码: {}, 错误响应: {}", response.statusCode(), errorBody);
                                        return Mono.error(new RuntimeException("第二次API请求失败: " + response.statusCode() + " - " + errorBody));
                                    });
                        })
                        .bodyToFlux(String.class)
                        .flatMap(chunk -> {
                            try {
                                String data = chunk.trim();
                                if (data.startsWith("data:")) data = data.substring(5).trim();
                                if (data.isEmpty() || "[DONE]".equals(data)) return Flux.empty();

                                JsonNode root = mapper.readTree(data);
                                JsonNode choice = root.path("choices").get(0);
                                String finishReason = choice.path("finish_reason").asText("");
                                
                                // 检测新的函数调用 - tool_calls格式
                                JsonNode toolCalls = choice.path("delta").path("tool_calls");
                                if (toolCalls.isMissingNode()) {
                                    toolCalls = choice.path("message").path("tool_calls");
                                }
                                
                                if (!toolCalls.isMissingNode() && toolCalls.isArray() && toolCalls.size() > 0) {
                                    JsonNode toolCall = toolCalls.get(0);
                                    JsonNode idNode = toolCall.path("id");
                                    if (!idNode.isMissingNode() && !idNode.isNull()) {
                                        String idChunk = idNode.asText("");
                                        if (!idChunk.isEmpty()) {
                                            nextToolCallIdRef.set(idChunk);
                                        }
                                    }
                                    
                                    JsonNode functionNode = toolCall.path("function");
                                    if (!functionNode.isMissingNode()) {
                                        JsonNode nameNode = functionNode.path("name");
                                        if (!nameNode.isMissingNode() && !nameNode.isNull()) {
                                            String nameChunk = nameNode.asText("");
                                            if (!nameChunk.isEmpty()) {
                                                nextFunctionNameRef.updateAndGet(current -> current + nameChunk);
                                            }
                                        }
                                        
                                        JsonNode argsNode = functionNode.path("arguments");
                                        if (!argsNode.isMissingNode() && !argsNode.isNull()) {
                                            String argsChunk = argsNode.asText("");
                                            if (!argsChunk.isEmpty()) {
                                                nextFunctionArgsRef.updateAndGet(current -> current + argsChunk);
                                            }
                                        }
                                    }
                                }
                                
                                // 检测新的函数调用 - function_call格式
                                JsonNode functionCall = choice.path("delta").path("function_call");
                                if (functionCall.isMissingNode()) {
                                    functionCall = choice.path("message").path("function_call");
                                }
                                
                                if (!functionCall.isMissingNode()) {
                                    JsonNode nameNode = functionCall.path("name");
                                    if (!nameNode.isMissingNode() && !nameNode.isNull()) {
                                        String nameChunk = nameNode.asText("");
                                        if (!nameChunk.isEmpty()) {
                                            nextFunctionNameRef.updateAndGet(current -> current + nameChunk);
                                        }
                                    }
                                    
                                    JsonNode argsNode = functionCall.path("arguments");
                                    if (!argsNode.isMissingNode() && !argsNode.isNull()) {
                                        String argsChunk = argsNode.asText("");
                                        if (!argsChunk.isEmpty()) {
                                            nextFunctionArgsRef.updateAndGet(current -> current + argsChunk);
                                        }
                                    }
                                }
                                
                                // 如果检测到新的函数调用完成，递归处理
                                if ("tool_calls".equals(finishReason) || "function_call".equals(finishReason)) {
                                    String nextName = nextFunctionNameRef.get();
                                    String nextArgs = nextFunctionArgsRef.get();
                                    String nextToolCallId = nextToolCallIdRef.get();
                                    
                                    if (!nextName.isEmpty()) {
                                        log.info("🔄 检测到新的函数调用需求，递归处理: {}", nextName);
                                        // 🔧 修复：继续使用同一个 aiBuffer，这样递归调用的结果会累积到同一个buffer
                                        return handleFunctionCall(nextName, nextArgs, nextToolCallId, newMessages, aiBuffer, sessionId, userId, depth + 1);
                                    }
                                }

                                // 正常流式内容输出
                                JsonNode contentNode = choice.path("delta").path("content");
                                if (contentNode.isMissingNode() || contentNode.isNull()) return Flux.empty();
                                
                                String content = contentNode.isNumber() 
                                        ? contentNode.asText() 
                                        : contentNode.asText("");
                                
                                if (content.isEmpty() || "null".equals(content)) return Flux.empty();

                                // 🚨 过滤错误的函数调用文本输出
                                if (content.contains("ロン") || 
                                    content.contains("<tool_call>") || 
                                    content.contains("</tool_call>") ||
                                    content.contains("send{") ||
                                    (content.contains("\"name\"") && content.contains("query_"))) {
                                    log.warn("⚠️ [火车查询] 检测到AI错误地在content中输出函数调用文本，已过滤: {}", content);
                                    return Flux.empty();
                                }

                                aiBuffer.append(content);
                                // 直接返回文本内容
                                return Flux.just(content);
                            } catch (Exception e) {
                                log.warn("解析流出错: {}", e.getMessage());
                                return Flux.empty();
                            }
                        })
                        .doOnComplete(() -> {
                            // 🔧 修复：移除这里的保存逻辑，统一由主流程保存，避免重复
                            log.info("🔍 [火车查询] doOnComplete 被触发 - depth={}, aiBuffer长度={}", depth, aiBuffer.length());
                            log.debug("ℹ️ [火车查询] 函数调用处理完成，AI回复将由主流程统一保存");
                        });
            }

            // ========== 处理天气查询 ==========
            if ("query_weather".equals(name)) {
                log.info("🌤️ 处理天气查询...");
                // 解析函数参数
                JsonNode args = mapper.readTree(arguments);
                String cityName = args.path("cityName").asText("");
                String extensions = args.path("extensions").asText("");
                if (extensions == null || extensions.isEmpty()) extensions = "all";

                log.info("🧠 AI 请求调用函数：query_weather({}, {})", cityName, extensions);

                // 调用天气服务查询天气信息
                String resultString;
                boolean isApiError = false;
                try {
                    log.info("⏳ 正在调用本地函数 weatherService.getWeatherByCityName()...");
                    List<String> weatherList = weatherService.getWeatherByCityName(cityName, extensions);
                    resultString = String.join("\n", weatherList);
                    log.info("✅ 天气查询成功！返回 {} 条信息", weatherList.size());
                    log.debug("📋 天气信息: {}", resultString);
                    
                    // 检查是否是API错误
                    // 🔧 修复：只在真正出错或无数据时标记错误
                    if (weatherList.isEmpty() || resultString.contains("超过") || resultString.contains("限制")) {
                        isApiError = true;
                        log.warn("⚠️ 天气API返回错误，将告知AI跳过此步骤");
                        resultString = "天气查询API暂时不可用（超出调用次数限制或请求频率限制），请跳过天气查询，继续其他步骤（如文章查询等），并基于现有信息为用户提供建议。";
                    }
                } catch (Exception e) {
                    log.error("❌ 调用weatherService失败: {}", e.getMessage(), e);
                    isApiError = true;
                    resultString = "天气查询服务异常，请跳过此步骤，继续其他查询。";
                }

                // ========== 构建完整的消息历史（包括函数调用和结果） ==========
                List<Map<String, Object>> newMessages = new ArrayList<>();
                // 复制历史消息
                for (Map<String, Object> msg : messages) {
                    newMessages.add(new HashMap<>(msg));
                }
                // 添加assistant的函数调用请求（根据模型类型选择格式）
                Map<String, Object> assistantMsg = new HashMap<>();
                assistantMsg.put("role", "assistant");
                
                if (model != null && model.contains("Qwen2-7B")) {
                    // Qwen2-7B 使用 function_call 格式
                    assistantMsg.put("function_call", Map.of(
                            "name", name,
                            "arguments", arguments
                    ));
                } else {
                    // Qwen2.5 使用 tool_calls 格式
                    Map<String, Object> toolCall = new HashMap<>();
                    toolCall.put("id", toolCallId.isEmpty() ? "call_" + System.currentTimeMillis() : toolCallId);
                    toolCall.put("type", "function");
                    toolCall.put("function", Map.of("name", name, "arguments", arguments));
                    assistantMsg.put("tool_calls", List.of(toolCall));
                }
                newMessages.add(assistantMsg);
                
                // 添加函数执行结果（根据模型类型选择格式）
                Map<String, Object> toolResult = new HashMap<>();
                if (model != null && model.contains("Qwen2-7B")) {
                    // Qwen2-7B 使用 function 角色
                    toolResult.put("role", "function");
                    toolResult.put("name", name);
                    toolResult.put("content", resultString);
                } else {
                    // Qwen2.5 使用 tool 角色
                    toolResult.put("role", "tool");
                    toolResult.put("content", resultString);
                    if (!toolCallId.isEmpty()) {
                        toolResult.put("tool_call_id", toolCallId);
                    }
                }
                newMessages.add(toolResult);

                // 第二次请求LLM（根据模型类型选择格式）
                Map<String, Object> trafficFunc = TrafficServiceImpl.functionSchema();
                Map<String, Object> weatherFunc = WeatherServiceImpl.functionSchema();
                Map<String, Object> secondBodyWeather = new HashMap<>();
                secondBodyWeather.put("model", model);
                secondBodyWeather.put("stream", true);
                secondBodyWeather.put("messages", newMessages);
                
                if (model != null && model.contains("Qwen2-7B")) {
                    // Qwen2-7B 使用 functions 格式
                    List<Map<String, Object>> functions = List.of(trafficFunc, weatherFunc);
                    secondBodyWeather.put("functions", functions);
                    secondBodyWeather.put("function_call", "auto");
                } else {
                    // Qwen2.5 使用 tools 格式
                    List<Map<String, Object>> toolsForWeather = List.of(
                            Map.of("type", "function", "function", trafficFunc),
                            Map.of("type", "function", "function", weatherFunc)
                    );
                    secondBodyWeather.put("tools", toolsForWeather);
                    secondBodyWeather.put("tool_choice", "auto");
                }

                // 发送第二次请求，让AI基于函数结果生成回复
                AtomicReference<String> nextFunctionNameRef2 = new AtomicReference<>("");
                AtomicReference<String> nextFunctionArgsRef2 = new AtomicReference<>("");
                AtomicReference<String> nextToolCallIdRef2 = new AtomicReference<>("");
                
                return siliconWebClient.post()
                        .uri("/chat/completions")
                        .bodyValue(secondBodyWeather)
                        .retrieve()
                        .onStatus(status -> status.is4xxClientError() || status.is5xxServerError(), response -> {
                            return response.bodyToMono(String.class)
                                    .flatMap(errorBody -> {
                                        log.error("❌ 第二次API请求失败 - 状态码: {}, 错误响应: {}", response.statusCode(), errorBody);
                                        return Mono.error(new RuntimeException("第二次API请求失败: " + response.statusCode() + " - " + errorBody));
                                    });
                        })
                        .bodyToFlux(String.class)
                        .flatMap(chunk -> {
                            try {
                                String data = chunk.trim();
                                if (data.startsWith("data:")) data = data.substring(5).trim();
                                if (data.isEmpty() || "[DONE]".equals(data)) return Flux.empty();

                                JsonNode root = mapper.readTree(data);
                                JsonNode choice = root.path("choices").get(0);
                                String finishReason = choice.path("finish_reason").asText("");
                                
                                // 检测新的函数调用 - tool_calls格式
                                JsonNode toolCalls = choice.path("delta").path("tool_calls");
                                if (toolCalls.isMissingNode()) {
                                    toolCalls = choice.path("message").path("tool_calls");
                                }
                                
                                if (!toolCalls.isMissingNode() && toolCalls.isArray() && toolCalls.size() > 0) {
                                    JsonNode toolCall = toolCalls.get(0);
                                    JsonNode idNode = toolCall.path("id");
                                    if (!idNode.isMissingNode() && !idNode.isNull()) {
                                        String idChunk = idNode.asText("");
                                        if (!idChunk.isEmpty()) {
                                            nextToolCallIdRef2.set(idChunk);
                                        }
                                    }
                                    
                                    JsonNode functionNode = toolCall.path("function");
                                    if (!functionNode.isMissingNode()) {
                                        JsonNode nameNode = functionNode.path("name");
                                        if (!nameNode.isMissingNode() && !nameNode.isNull()) {
                                            String nameChunk = nameNode.asText("");
                                            if (!nameChunk.isEmpty()) {
                                                nextFunctionNameRef2.updateAndGet(current -> current + nameChunk);
                                            }
                                        }
                                        
                                        JsonNode argsNode = functionNode.path("arguments");
                                        if (!argsNode.isMissingNode() && !argsNode.isNull()) {
                                            String argsChunk = argsNode.asText("");
                                            if (!argsChunk.isEmpty()) {
                                                nextFunctionArgsRef2.updateAndGet(current -> current + argsChunk);
                                            }
                                        }
                                    }
                                }
                                
                                // 检测新的函数调用 - function_call格式
                                JsonNode functionCall = choice.path("delta").path("function_call");
                                if (functionCall.isMissingNode()) {
                                    functionCall = choice.path("message").path("function_call");
                                }
                                
                                if (!functionCall.isMissingNode()) {
                                    JsonNode nameNode = functionCall.path("name");
                                    if (!nameNode.isMissingNode() && !nameNode.isNull()) {
                                        String nameChunk = nameNode.asText("");
                                        if (!nameChunk.isEmpty()) {
                                            nextFunctionNameRef2.updateAndGet(current -> current + nameChunk);
                                        }
                                    }
                                    
                                    JsonNode argsNode = functionCall.path("arguments");
                                    if (!argsNode.isMissingNode() && !argsNode.isNull()) {
                                        String argsChunk = argsNode.asText("");
                                        if (!argsChunk.isEmpty()) {
                                            nextFunctionArgsRef2.updateAndGet(current -> current + argsChunk);
                                        }
                                    }
                                }
                                
                                // 如果检测到新的函数调用完成，递归处理
                                if ("tool_calls".equals(finishReason) || "function_call".equals(finishReason)) {
                                    String nextName = nextFunctionNameRef2.get();
                                    String nextArgs = nextFunctionArgsRef2.get();
                                    String nextToolCallId = nextToolCallIdRef2.get();
                                    
                                    if (!nextName.isEmpty()) {
                                        log.info("🔄 检测到新的函数调用需求，递归处理: {}", nextName);
                                        // 🔧 修复：继续使用同一个 aiBuffer，这样递归调用的结果会累积到同一个buffer
                                        return handleFunctionCall(nextName, nextArgs, nextToolCallId, newMessages, aiBuffer, sessionId, userId, depth + 1);
                                    }
                                }

                                // 正常流式内容输出
                                JsonNode contentNode = choice.path("delta").path("content");
                                if (contentNode.isMissingNode() || contentNode.isNull()) return Flux.empty();
                                
                                String content = contentNode.isNumber() 
                                        ? contentNode.asText() 
                                        : contentNode.asText("");
                                
                                if (content.isEmpty() || "null".equals(content)) return Flux.empty();

                                // 🚨 过滤错误的函数调用文本输出
                                if (content.contains("ロン") || 
                                    content.contains("<tool_call>") || 
                                    content.contains("</tool_call>") ||
                                    content.contains("send{") ||
                                    (content.contains("\"name\"") && content.contains("query_"))) {
                                    log.warn("⚠️ [天气查询] 检测到AI错误地在content中输出函数调用文本，已过滤: {}", content);
                                    return Flux.empty();
                                }

                                aiBuffer.append(content);
                                // 直接返回文本内容
                                return Flux.just(content);
                            } catch (Exception e) {
                                log.warn("解析流出错: {}", e.getMessage());
                                return Flux.empty();
                            }
                        })
                        .doOnComplete(() -> {
                            // 🔧 修复：移除这里的保存逻辑，统一由主流程保存，避免重复
                            log.info("🔍 [天气查询] doOnComplete 被触发 - depth={}, aiBuffer长度={}", depth, aiBuffer.length());
                            log.debug("ℹ️ [天气查询] 函数调用处理完成，AI回复将由主流程统一保存");
                        });
            }

            // ========== 处理向量数据库查询 ==========
            if ("query_qdrant".equals(name)) {
                log.info("📚 处理向量数据库查询...");
                // 解析函数参数
                JsonNode args = mapper.readTree(arguments);
                String text = args.path("text").asText("");

                log.info("🧠 AI 请求调用函数：query_qdrant({})", text);

                // 🔧 修复：创建 final 变量副本，供 lambda 表达式使用
                final String finalName = name;
                final String finalArguments = arguments;
                final String finalToolCallId = toolCallId;
                
                // 🔧 修复：使用响应式方式调用 SearchByText，避免阻塞
                log.info("⏳ 正在调用本地函数 qdrantService.SearchByText()...");
                
                return qdrantService.SearchByText(text)
                        .flatMapMany(essayList -> {
                            // 将查询结果格式化为 Markdown
                            String resultString;
                            if (essayList == null || essayList.isEmpty()) {
                                log.info("⚠️ 未找到相关文章");
                                resultString = "未找到相关文章。";
                            } else {
                                log.info("✅ 向量查询成功！找到 {} 篇相关文章", essayList.size());
                                StringBuilder markdownBuilder = new StringBuilder();
                                markdownBuilder.append("找到以下相关文章：\n\n");
                                
                                for (EssayDTO essay : essayList) {
                                    // 文章标题（带超链接，链接以 id 形式拼接，前缀先空着）
                                    String title = essay.getTitle() != null ? essay.getTitle() : "无标题";
                                    Integer eId = essay.getEId();
                                    String link = eId != null ? "/" + eId : "#";
                                    markdownBuilder.append("## [").append(title).append("](").append(link).append(")\n\n");
                                    
                                    // 文章封面图片（取第一张图片）
                                    List<String> images = essay.getImages();
                                    if (images != null && !images.isEmpty()) {
                                        String coverImage = images.get(0);
                                        if (coverImage != null && !coverImage.trim().isEmpty()) {
                                            markdownBuilder.append("![封面](").append(coverImage).append(")\n\n");
                                        }
                                    }
                                }
                                
                                resultString = markdownBuilder.toString();
                                log.debug("📄 Markdown结果: {}", resultString.substring(0, Math.min(200, resultString.length())) + "...");
                            }
                            
                            // ========== 构建完整的消息历史（包括函数调用和结果） ==========
                            List<Map<String, Object>> newMessages = new ArrayList<>();
                            // 复制历史消息
                            for (Map<String, Object> msg : messages) {
                                newMessages.add(new HashMap<>(msg));
                            }
                            // 添加assistant的函数调用请求（根据模型类型选择格式）
                            Map<String, Object> assistantMsg = new HashMap<>();
                            assistantMsg.put("role", "assistant");
                            
                            if (model != null && model.contains("Qwen2-7B")) {
                                // Qwen2-7B 使用 function_call 格式
                                assistantMsg.put("function_call", Map.of(
                                        "name", finalName,
                                        "arguments", finalArguments
                                ));
                            } else {
                                // Qwen2.5 使用 tool_calls 格式
                                Map<String, Object> toolCall = new HashMap<>();
                                toolCall.put("id", finalToolCallId.isEmpty() ? "call_" + System.currentTimeMillis() : finalToolCallId);
                                toolCall.put("type", "function");
                                toolCall.put("function", Map.of("name", finalName, "arguments", finalArguments));
                                assistantMsg.put("tool_calls", List.of(toolCall));
                            }
                            newMessages.add(assistantMsg);
                            
                            // 添加函数执行结果（根据模型类型选择格式）
                            Map<String, Object> toolResult = new HashMap<>();
                            if (model != null && model.contains("Qwen2-7B")) {
                                // Qwen2-7B 使用 function 角色
                                toolResult.put("role", "function");
                                toolResult.put("name", finalName);
                                toolResult.put("content", resultString);
                            } else {
                                // Qwen2.5 使用 tool 角色
                                toolResult.put("role", "tool");
                                toolResult.put("content", resultString);
                                if (!finalToolCallId.isEmpty()) {
                                    toolResult.put("tool_call_id", finalToolCallId);
                                }
                            }
                            newMessages.add(toolResult);
                            
                            // 🔧 新增：添加指示AI直接输出Markdown的消息
                            newMessages.add(Map.of(
                                "role", "system",
                                "content", "请直接输出上面的Markdown格式内容，不要做任何改写或总结。保持原始的Markdown格式，包括标题、链接和图片。"
                            ));

                            // 第二次请求LLM（根据模型类型选择格式）
                            Map<String, Object> trafficFunc = TrafficServiceImpl.functionSchema();
                            Map<String, Object> weatherFunc = WeatherServiceImpl.functionSchema();
                            Map<String, Object> vectorFunc = QdrantVectorService.functionSchema();
                            Map<String, Object> secondBodyQdrant = new HashMap<>();
                            secondBodyQdrant.put("model", model);
                            secondBodyQdrant.put("stream", true);
                            secondBodyQdrant.put("messages", newMessages);
                            
                            if (model != null && model.contains("Qwen2-7B")) {
                                // Qwen2-7B 使用 functions 格式
                                List<Map<String, Object>> functions = List.of(trafficFunc, weatherFunc, vectorFunc);
                                secondBodyQdrant.put("functions", functions);
                                secondBodyQdrant.put("function_call", "auto");
                            } else {
                                // Qwen2.5 使用 tools 格式
                                List<Map<String, Object>> toolsForQdrant = List.of(
                                        Map.of("type", "function", "function", trafficFunc),
                                        Map.of("type", "function", "function", weatherFunc),
                                        Map.of("type", "function", "function", vectorFunc)
                                );
                                secondBodyQdrant.put("tools", toolsForQdrant);
                                secondBodyQdrant.put("tool_choice", "auto");
                            }

                            // 发送第二次请求，让AI基于函数结果生成回复
                            AtomicReference<String> nextFunctionNameRef3 = new AtomicReference<>("");
                            AtomicReference<String> nextFunctionArgsRef3 = new AtomicReference<>("");
                            AtomicReference<String> nextToolCallIdRef3 = new AtomicReference<>("");
                            
                            return siliconWebClient.post()
                                    .uri("/chat/completions")
                                    .bodyValue(secondBodyQdrant)
                                    .retrieve()
                                    .onStatus(status -> status.is4xxClientError() || status.is5xxServerError(), response -> {
                                        return response.bodyToMono(String.class)
                                                .flatMap(errorBody -> {
                                                    log.error("❌ 第二次API请求失败 - 状态码: {}, 错误响应: {}", response.statusCode(), errorBody);
                                                    return Mono.error(new RuntimeException("第二次API请求失败: " + response.statusCode() + " - " + errorBody));
                                                });
                                    })
                                    .bodyToFlux(String.class)
                                    .flatMap(chunk -> {
                                        try {
                                            String data = chunk.trim();
                                            if (data.startsWith("data:")) data = data.substring(5).trim();
                                            if (data.isEmpty() || "[DONE]".equals(data)) return Flux.empty();

                                            JsonNode root = mapper.readTree(data);
                                            JsonNode choice = root.path("choices").get(0);
                                            String finishReason = choice.path("finish_reason").asText("");
                                            
                                            // 检测新的函数调用 - tool_calls格式
                                            JsonNode toolCalls = choice.path("delta").path("tool_calls");
                                            if (toolCalls.isMissingNode()) {
                                                toolCalls = choice.path("message").path("tool_calls");
                                            }
                                            
                                            if (!toolCalls.isMissingNode() && toolCalls.isArray() && toolCalls.size() > 0) {
                                                JsonNode toolCall = toolCalls.get(0);
                                                JsonNode idNode = toolCall.path("id");
                                                if (!idNode.isMissingNode() && !idNode.isNull()) {
                                                    String idChunk = idNode.asText("");
                                                    if (!idChunk.isEmpty()) {
                                                        nextToolCallIdRef3.set(idChunk);
                                                    }
                                                }
                                                
                                                JsonNode functionNode = toolCall.path("function");
                                                if (!functionNode.isMissingNode()) {
                                                    JsonNode nameNode = functionNode.path("name");
                                                    if (!nameNode.isMissingNode() && !nameNode.isNull()) {
                                                        String nameChunk = nameNode.asText("");
                                                        if (!nameChunk.isEmpty()) {
                                                            nextFunctionNameRef3.updateAndGet(current -> current + nameChunk);
                                                        }
                                                    }
                                                    
                                                    JsonNode argsNode = functionNode.path("arguments");
                                                    if (!argsNode.isMissingNode() && !argsNode.isNull()) {
                                                        String argsChunk = argsNode.asText("");
                                                        if (!argsChunk.isEmpty()) {
                                                            nextFunctionArgsRef3.updateAndGet(current -> current + argsChunk);
                                                        }
                                                    }
                                                }
                                            }
                                            
                                            // 检测新的函数调用 - function_call格式
                                            JsonNode functionCall = choice.path("delta").path("function_call");
                                            if (functionCall.isMissingNode()) {
                                                functionCall = choice.path("message").path("function_call");
                                            }
                                            
                                            if (!functionCall.isMissingNode()) {
                                                JsonNode nameNode = functionCall.path("name");
                                                if (!nameNode.isMissingNode() && !nameNode.isNull()) {
                                                    String nameChunk = nameNode.asText("");
                                                    if (!nameChunk.isEmpty()) {
                                                        nextFunctionNameRef3.updateAndGet(current -> current + nameChunk);
                                                    }
                                                }
                                                
                                                JsonNode argsNode = functionCall.path("arguments");
                                                if (!argsNode.isMissingNode() && !argsNode.isNull()) {
                                                    String argsChunk = argsNode.asText("");
                                                    if (!argsChunk.isEmpty()) {
                                                        nextFunctionArgsRef3.updateAndGet(current -> current + argsChunk);
                                                    }
                                                }
                                            }
                                            
                                            // 如果检测到新的函数调用完成，递归处理
                                            if ("tool_calls".equals(finishReason) || "function_call".equals(finishReason)) {
                                                String nextName = nextFunctionNameRef3.get();
                                                String nextArgs = nextFunctionArgsRef3.get();
                                                String nextToolCallId = nextToolCallIdRef3.get();
                                                
                                                if (!nextName.isEmpty()) {
                                                    log.info("🔄 检测到新的函数调用需求，递归处理: {}", nextName);
                                                    // 🔧 修复：继续使用同一个 aiBuffer，这样递归调用的结果会累积到同一个buffer
                                                    return handleFunctionCall(nextName, nextArgs, nextToolCallId, newMessages, aiBuffer, sessionId, userId, depth + 1);
                                                }
                                            }

                                            // 正常流式内容输出
                                            JsonNode contentNode = choice.path("delta").path("content");
                                            if (contentNode.isMissingNode() || contentNode.isNull()) return Flux.empty();
                                            
                                            String content = contentNode.isNumber() 
                                                    ? contentNode.asText() 
                                                    : contentNode.asText("");
                                            
                                            if (content.isEmpty() || "null".equals(content)) return Flux.empty();

                                            // 🚨 过滤错误的函数调用文本输出
                                            if (content.contains("ロン") || 
                                                content.contains("<tool_call>") || 
                                                content.contains("</tool_call>") ||
                                                content.contains("send{") ||
                                                (content.contains("\"name\"") && content.contains("query_"))) {
                                                log.warn("⚠️ [向量查询] 检测到AI错误地在content中输出函数调用文本，已过滤: {}", content);
                                                return Flux.empty();
                                            }

                                            aiBuffer.append(content);
                                            // 直接返回文本内容
                                            return Flux.just(content);
                                        } catch (Exception e) {
                                            log.warn("解析流出错: {}", e.getMessage());
                                            return Flux.empty();
                                        }
                                    })
                                    .doOnComplete(() -> {
                                        // 🔧 修复：移除这里的保存逻辑，统一由主流程保存，避免重复
                                        log.info("🔍 [向量查询] doOnComplete 被触发 - depth={}, aiBuffer长度={}", depth, aiBuffer.length());
                                        log.debug("ℹ️ [向量查询] 函数调用处理完成，AI回复将由主流程统一保存");
                                    });
                        }); // 闭合 flatMapMany
            }

            // ========== 未知函数调用 ==========
            log.warn("⚠️ 未知的函数调用: {}", name);
            return Flux.just("未知的函数调用: " + name);
        } catch (Exception e) {
            // 函数调用处理失败
            log.error("处理函数调用失败: {}", e.getMessage(), e);
            return Flux.just("处理函数调用出错: " + e.getMessage());
        }
    }
}